/*
 * @Author: 李石
 * @Date: 2024-05-21 08:12:32
 * @LastEditors: lishi
 * @LastEditTime: 2024-07-18 17:26:08
 * @Description: 
 * Copyright (c) 2024 by ${lishi0105@163.com}, All Rights Reserved. 
 */
#include <cmath>
#include <string.h>
#include "logger.hpp"
#include "utilspp.hpp"
#include "FramePusher.hpp"
#include "ZLMFramePuller.hpp"
#include "StreamProxyer.hpp"

int StreamProxyer::m_server_port = 0;
std::string StreamProxyer::m_local_ip = "127.0.0.1";

static bool isRtspStream(const std::string &url)
{
    // do nothing
    if(strstr(url.c_str(), "rtsp://") != nullptr){
        return true;
    }
    return false;
}

static std::string urlEncode(const std::string& input) {
    std::ostringstream encoded;
    encoded.fill('0');
    encoded << std::hex;

    for (char ch : input) {
        // Keep alphanumeric and other accepted characters intact
        if (isalnum(static_cast<unsigned char>(ch)) || ch == '-' || ch == '.' || ch == '~') {
            encoded << ch;
        }
        // Any other characters are percent-encoded
        else {
            encoded << '%' << std::setw(2) << static_cast<int>(static_cast<unsigned char>(ch));
        }
    }

    return encoded.str();
}

StreamProxyer::StreamProxyer(const std::string &app, 
                             const std::string &stream, 
                             const std::string &url, 
                             int32_t rtp_type):
    m_rtp_type(rtp_type),
    m_push_cnt(0),
    m_codec_id(0),
    m_proxy_alive(false),
    m_proxy_exit(false),
    m_play_url(""),
    m_src_url(url),
    m_app_name(app),
    m_stream_name(stream)
{
    m_play_url = UtilsPP::formatString("http://%s:%d/%s/%s.live.flv", 
                                    StreamProxyer::m_local_ip.c_str(), 
                                    StreamProxyer::m_server_port, 
                                    m_app_name.c_str(), 
                                    m_stream_name.c_str());
    LOGI("play: {}", m_play_url);
    m_monitor_thread = std::thread(&StreamProxyer::monitorThread, this);
    m_push_thread = std::thread(&StreamProxyer::pushFrameThread, this);
}

StreamProxyer::~StreamProxyer()
{
    m_proxy_exit.store(true);
    if(m_monitor_thread.joinable()){
        m_monitor_thread.join();
    }
    if(m_push_thread.joinable()){
        m_push_thread.join();
    }
    LOGI("StreamProxyer destruct");
}

void StreamProxyer::pushFrameThread()
{
    while(!m_proxy_exit.load()){
        bool m_pusher_exit = false;
        while(!m_proxy_exit.load() && !m_pusher_exit){
            m_codec_id.store(-1);
            FramePusher::Ptr m_stream_pusher = nullptr;
            std::atomic<bool> m_find_idr(false);
            auto data_callback = [this, m_stream_pusher, &m_find_idr](const FrameData::Ptr data){
                if(data == nullptr)return;
                if(m_proxy_exit.load())return;
                {
                    std::lock_guard<std::mutex> lock(m_pusher_mtx);
                    if(m_stream_pusher.get() == nullptr){
                        return ;
                    }
                    if(!m_find_idr.load()){
                        if(!data->configFrame() && !data->keyFrame()) return;
                        m_find_idr.store(true);
                    }
                    m_stream_pusher->PushFrameData(data);
                }
                
            };
            std::string encoded_url = urlEncode(m_src_url);
            auto m_stream_puller = std::make_shared<ZLMFramePuller>(encoded_url, m_rtp_type, data_callback);
            if(m_stream_puller == nullptr){
                std::runtime_error("stream puller is nullptr");
            }
            int32_t counter = 0;
            while(!m_proxy_exit.load() && !m_pusher_exit){
                counter ++;
                if(counter > 50){
                    m_pusher_exit = true;
                    break;
                }
                std::this_thread::sleep_for(std::chrono::milliseconds(100));
                if(!m_stream_puller->alive()){
                    LOGE("stream puller not alive");
                    continue;
                }
                if(m_stream_puller->codec_id() != 0 && m_stream_puller->codec_id() != 1){
                    LOGE("stream puller codec_id is invalid");
                    continue;;
                }
            }
            if(m_pusher_exit){
                break;
            }
            auto codec_id = m_stream_puller->codec_id();
            m_codec_id.store(codec_id);
            m_stream_pusher = FramePusher::CreateShared(m_app_name, m_stream_name, codec_id);
            if(m_stream_pusher.get() == nullptr){
                LOGE("create stream pusher error");
                m_pusher_exit = true;
                break;
            }
            while(!m_proxy_exit.load() && !m_pusher_exit){
                std::this_thread::sleep_for(std::chrono::seconds(1));
                bool bAlive = m_stream_puller->alive();
                auto int32Codec_id = m_stream_puller->codec_id();
                if(!bAlive || int32Codec_id != codec_id){
                    LOGE("stream puller not alive: {} or codec_id {}->{}", bAlive, codec_id, codec_id);
                    m_pusher_exit = true;
                }
            }
        }
        LOGI("restart pusher ....");
    }
    LOGW("pusher thread exit");
}

void StreamProxyer::monitorThread()
{
    LOGI("monitorThread start");
    int32_t m_push_cnt_save = 0;
    while(!m_proxy_exit.load()){
        std::this_thread::sleep_for(std::chrono::seconds(3));
        if(m_push_cnt != m_push_cnt_save){
            m_push_cnt_save = m_push_cnt;
            if(!m_proxy_alive.load()){
                m_proxy_alive.store(true);
            }
            continue;
        }
        m_proxy_alive.store(false);
    }
    m_proxy_alive.store(false);
    LOGI("monitorThread exit");
}